package com.gper.edu.flink.stream.window;

import com.gper.edu.flink.stream.StreamEnvironment;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.evictors.TimeEvictor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.triggers.ContinuousEventTimeTrigger;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

/**
 * @Author: ellis.guan
 * @Description:
 * @Date: 2020/11/8 9:51
 * @Version: 1.0
 */
public class WindowTriggerDemo extends StreamEnvironment {

    public static void main(String[] args) throws Exception {
        WindowTriggerDemo triggerDemo = new WindowTriggerDemo();
        triggerDemo.runJob();
    }

    public void execute() throws Exception {
        setParallelism(1);
        DataStreamSource<String> source = getDataStreamSource();
        SingleOutputStreamOperator<Tuple3<Integer, Double, Long>> flatMapStream = source.flatMap(new FlatMapFunction<String, Tuple3<Integer, Double, Long>>() {
            @Override
            public void flatMap(String s, Collector<Tuple3<Integer, Double, Long>> collector) throws Exception {
                String[] vaules = s.split(",");
                Tuple3<Integer, Double, Long> returnValue = null;
                try {
                    if (vaules.length < 3) return;
                    returnValue = Tuple3.of(Integer.valueOf(vaules[0]), Double.valueOf(vaules[1]), new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").parse(vaules[2]).getTime());
                } catch (Exception e) {

                }
                if (returnValue != null)
                    collector.collect(returnValue);
            }
        });

//        flatMapStream.print("--->");

        flatMapStream
                .assignTimestampsAndWatermarks(new WindowTriggerWatermark(Time.seconds(30)))
                .keyBy(new KeySelector<Tuple3<Integer, Double, Long>, Integer>() {
                    @Override
                    public Integer getKey(Tuple3<Integer, Double, Long> value) throws Exception {
                        return value.f0;
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.minutes(10)))
                .allowedLateness(Time.minutes(2))
                .trigger(ContinuousEventTimeTrigger.of(Time.seconds(30)))  // 每30秒触发一次计算
                .evictor(TimeEvictor.of(Time.minutes(0), true))
                // true : 计算后进行元素驱除，驱除的策略 element 满足如下条件： element.timestamp <= elements.maxTimestamp - TimeEvictor.timestamp的时间截的元素
                .aggregate(new MyAggregateFunction(),new MyWindowSum()).print("===========");
                // 如果使用了驱除策略，每一次只累加每次触发的元素集合（30秒内的数据，并不会在上一次30秒的累加结果上进行累加计算），而不是整个窗口的数据
        env.execute("WindowTriggerDemo");
    }
}
